package ssx.demo.hbase2demo.service;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;

@Service("myHbaseService")
public class SerDemo {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired
    @Qualifier("myConnection")
    private Connection hbaseConnection;

    String tableName = "hello_ssx_tab";
    String familyName = "family1";

    /**
     * 查询value一般情况下通过rowkey最快了
     * 如果使用列作为条件查询就很慢 因为hbase默认只有rowkey索引
     * 但是也支持列过滤查询 类似于oracle的where条件 但是查询慢
     *
     */
    public void queryByCondition() throws IOException {
        Scan scan = new Scan();
        //指定value值
        ValueFilter valueFilter = new ValueFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("单元格值")));
        //指定行键
        RowFilter rowFilter = new RowFilter(CompareOperator.LESS_OR_EQUAL, new RegexStringComparator("001"));
        //也可以指定列族名称 列名称 行键 值等
        //比较是否相等 大于等于小于 支持正则 值等方式
        ArrayList<Filter> list = new ArrayList<>();
        list.add(valueFilter);
        list.add(rowFilter);

        FilterList filterList = new FilterList();
        filterList.addFilter(list);

        scan.setFilter(filterList);

        ResultScanner scanner = hbaseConnection.getTable(TableName.valueOf("表名")).getScanner(scan);
        Iterator<Result> iterator = scanner.iterator();
        while (iterator.hasNext()){
            Result next = iterator.next();
            Cell[] cells = next.rawCells();
            for (Cell cell : cells) {
                String row = Bytes.toString(CellUtil.cloneRow(cell));
                String f = Bytes.toString(CellUtil.cloneFamily(cell));
                String v = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println("输出的结果 row+f+v");
            }
        }

    }


    /**
     * 在hbase创建一张表，只指定表名、列族名。其他配置均默认
     * 表名是tableName = "hello_ssx_tab";
     * 列族名字是familyName = "family1";
     * @return
     */
    public boolean addTab(){
        try {
            // logger.info("获取hbaseAdmin对象用于操作表");
            Admin hbaseConnectionAdmin = hbaseConnection.getAdmin();

            //判断表是否存在
            TableName tableName1 = TableName.valueOf(this.tableName);
            boolean tableExists = hbaseConnectionAdmin.tableExists(tableName1);
            logger.info("表是否存在"+tableName+"："+tableExists);

            //如果表不存在就新建
            if (!tableExists){
                //表不存在
                //表 描述建造者
                TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName1);
                //列族 描述建造者
                ColumnFamilyDescriptorBuilder cfdb = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(familyName));

                //设置最大版本号 其实就是记录的轨迹信息，设置为3可以查询追溯到前1个版本前2个版本前三个版本，此功能相当于数据库的flashback闪回。默认1
                cfdb.setMaxVersions(3);

                //创建列族 描述
                ColumnFamilyDescriptor familyDescriptor = cfdb.build();
                //将列族加入到表描述中
                builder.setColumnFamily(familyDescriptor);
                //创建表 描述
                TableDescriptor tableDescriptor = builder.build();

                //创建表
                hbaseConnectionAdmin.createTable(tableDescriptor);
                 logger.info("表创建成功"+tableName);

            }else {
                logger.warn("表已经存在："+tableName);
            }

            return hbaseConnectionAdmin.tableExists(tableName1);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 向表"hello_ssx_tab";中的列族"family1";添加一条记录
     *             String colName = "col_a";
     *             String rowKey = "row_001";
     * @param value
     * @return
     */
    public boolean setValue(String value,String colName,String rowKey){
        //指定表名
        TableName tableName1 = TableName.valueOf(this.tableName);
        try {

            //获取表对象
            Table table = hbaseConnection.getTable(tableName1);
            //设置行键值
            Put row001 = new Put(Bytes.toBytes(rowKey));


            //设置列族，列，值
            row001.addColumn(Bytes.toBytes(familyName),
                    Bytes.toBytes(colName),Bytes.toBytes(value));




            //添加记录
            table.put(row001);
            // logger.info("成功hbase添加数据信息：目标表："+tableName+",目标行键："+rowKey+",目标列族："+familyName+",目标列："+ colName +",目标值："+value);
            return true;

        } catch (IOException e) {
            e.printStackTrace();
        }

        return false;

    }

    /**
     * 批量插入、读取、删除、修改
     *  param value 值
     *  param colName 列名
     *  param rowKey 行键
     * @return
     */
    public boolean setValues() throws IOException {

        //指定表名
        TableName tableName1 = TableName.valueOf(this.tableName);
        try {


            //设置行键值- 待添加的数据
            Put put = new Put(Bytes.toBytes("rowKey001")) //设置行键
                            .addColumn(Bytes.toBytes(familyName), //设置列族
                                       Bytes.toBytes("colName01"), //设置列
                                       Bytes.toBytes("value01")); //设置值
            put.addColumn(Bytes.toBytes(familyName), //设置列族
                                       Bytes.toBytes("colName02"), //设置列
                                       Bytes.toBytes("value02")); //设置值
            //...

            //设置待查询的数据
            Get get1 = new Get(Bytes.toBytes("rowKey002")).addFamily(Bytes.toBytes(familyName));
            Get get2 = new Get(Bytes.toBytes("rowKey003")).addFamily(Bytes.toBytes(familyName));
            Get get3 = new Get(Bytes.toBytes("rowKey004")).addFamily(Bytes.toBytes(familyName));
            //...

            //设置删除数据
            Delete delete = new Delete(Bytes.toBytes("rowKey005")).addFamily(Bytes.toBytes(familyName));

            //方法1：批量提交 简单实现 start
            {
                Table table = hbaseConnection.getTable(tableName1);
                ArrayList<Put> puts = new ArrayList<>();
                ArrayList<Get> gets = new ArrayList<>();
                ArrayList<Delete> deletes = new ArrayList<>();
                table.put(puts);
                Result[] results = table.get(gets);
                for (Result result : results) {
                    Cell[] cells = result.rawCells();
                    for (Cell cell : cells) {
                        byte[] bytes = CellUtil.cloneValue(cell);
                        System.out.println("值：" + Bytes.toString(bytes));
                    }
                }
                table.delete(deletes);
            }
            //方法1：批量提交 简单实现 end



            //方法2：批量提交 通过线程池 高度自定义 好像不支持查询
            //下面是创建缓存、批量提交到hbase 可以自定义线程池
            {
                BufferedMutator bufferedMutator = hbaseConnection.getBufferedMutator(tableName1);
                BufferedMutator bufferedMutator2 = hbaseConnection.getBufferedMutator(new BufferedMutatorParams(tableName1).pool(null));

                ArrayList<Mutation> mutations = new ArrayList<>();
                mutations.add(put);
                mutations.add(delete);
                bufferedMutator.mutate(mutations);
            }






            //添加记录
             // logger.info("成功hbase添加数据信息：目标表："+tableName+",目标行键："+rowKey+",目标列族："+familyName+",目标列："+ colName +",目标值："+value);
            return true;

        } catch (IOException e) {
            e.printStackTrace();
        }

        return false;

    }

    /**
     * 查询表中的记录 指定rowkey
     * String rowKey = "row_001";
     * @param rowKey
     * @return
     */
    public String getValue(String rowKey){
        //指定表名
        TableName tableName1 = TableName.valueOf(this.tableName);
        try {
            //设置表
            Table table = hbaseConnection.getTable(tableName1);
            //指定要查询的行键
            Get get = new Get(Bytes.toBytes(rowKey));
            //查询返回结果
            Result result = table.get(get);
            //获取结果集合
            Cell[] cells = result.rawCells();
            //遍历结果集
            StringBuilder returnData = new StringBuilder();
            for (Cell cell:cells){
                byte[] bytes = CellUtil.cloneRow(cell);
                String s = Bytes.toString(bytes);
                 logger.info("查询hbase数据信息：目标表："+tableName+",目标行键："+rowKey+",查询出的信息： 行键："+s);
                returnData.append(s).append('\n');
                String s1 = Bytes.toString(CellUtil.cloneFamily(cell));
                 logger.info("查询hbase数据信息：目标表："+tableName+",目标行键："+rowKey+",查询出的信息： 列族："+s1);
                returnData.append(s1).append('\n');
                String s2 = Bytes.toString(CellUtil.cloneQualifier(cell));
                 logger.info("查询hbase数据信息：目标表："+tableName+",目标行键："+rowKey+",查询出的信息： 列名："+s2);
                returnData.append(s2).append('\n');
                String s3 = Bytes.toString(CellUtil.cloneValue(cell));
                 logger.info("查询hbase数据信息：目标表："+tableName+",目标行键："+rowKey+",查询出的信息： 值："+s3);
                returnData.append(s3).append('\n');
            }
            return returnData.toString();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return "查询失败";
    }

}
